package day02

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
  * 访问次数最多用户前5个
  */
object TOPN {
  def main(args: Array[String]): Unit = {
    //初始化
    val sparkConf: SparkConf = new SparkConf().setAppName("PV").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    //加载文件
    val data: RDD[String] = sc.textFile("file:///F:\\weblog\\input\\access.log.20181101.dat");
  //切分数据拿到所有的ip
    val ips: RDD[String] = data.map(x=>x.split(" ")(0))
    //将ip地址组成元组
    val ipAnd1: RDD[(String, Int)] = ips.map(x=>(x,1))
  //统计ip数
    val resultRDD: RDD[(String, Int)] = ipAnd1.reduceByKey(_+_)
    //排序
    val sortRDD: RDD[(String, Int)] = resultRDD.sortBy(_._2,false)
    //获取前5个
    val result: Array[(String, Int)] = sortRDD.take(5)
    //广播变量
    val broadcast: Broadcast[Array[(String, Int)]] = sc.broadcast(result)
    //获取广播变量的值  广播变量解决的问题，防止每个task都加载一份数据，如果当前进程
    //中task比较多，有几个task就会加载几分，这样会占用过多的内存，为了解决这样的问题，就有了广播变量
    val value: Array[(String, Int)] = broadcast.value

    //输出结果
  println(result.toBuffer)
    //释放资源
    sc.stop()
  }
}
